package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @auther yuzehua
 * @date 2022/10/25 - 19:51
 */
public class Dwd_PlayUserChapterProcess extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_PlayUserChapterProcess().init(
                2002,
                2,
                "Dwd_PlayUserChapterProcess"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_PlayUserChapterProcess");
        // 2. 过滤出用户章节进度表（user_chapter_process）
        Table user_chapter_process = tEnv.sqlQuery("select " +
                " `data`['id'] id," +
                " `data`['chapter_id'] chapter_id," +
                " `data`['user_id'] user_id," +
                " `data`['position_sec'] position_sec," +
                " ts," +
                " pt" +
                " from ods_db" +
                " where `database` = 'edu'" +
                " and `table` = 'user_chapter_process'" +
                " and `type` = 'insert'");
        tEnv.createTemporaryView("user_chapter_process", user_chapter_process);
        Table result = tEnv.sqlQuery("select * from user_chapter_process");
        //result.execute().print();

        // 3. 写出到kafka中
        tEnv.executeSql("create table dwd_play_user_chapter_process(" +
                " id string," +
                " chapter_id string," +
                " user_id string," +
                " position_sec string," +
                " ts bigint," +
                " pt TIMESTAMP_LTZ(3)" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_PLAY_USER_CHAPTER_PROCESS));

        result.executeInsert("dwd_play_user_chapter_process");


    }
}
